package com.darcytech.debezium.core.pool;

import com.darcytech.debezium.common.utils.StringLockUtils;
import com.darcytech.debezium.core.work.RegisterExecutor;
import com.darcytech.debezium.core.work.UserAndPassHelper;
import com.google.common.collect.Queues;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public final class ConnectorConnectionBinding implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConnectorConnectionBinding.class);

    static {
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
        } catch (ClassNotFoundException ignored) {

        }
    }

    private static final String EXPRESSION_PREFIX = "${";
    private static final Long DEFAULT_DETECTION_MS = 30000L;
    private static final Long REGISTER_FAIL_RETRY_INTERVAL_TIME = 30000L;
    private static final Integer DEFAULT_GROUP_AVERAGE_NUMBER = 10;
    private static final Long DEFAULT_REGISTER_MAX_DELAY_SECONDS = 5L;
    private static final AtomicInteger threadNum = new AtomicInteger();

    private final String kafkaServers;

    private final ConcurrentMap<String, MetaData> metaDataMap;
    private final BlockingQueue<String> doRegisterEvent;
    private final Long connectionDetectionTime;
    private final Long registerFailRetryIntervalTime;
    private final Integer groupAverageNumber;
    private final Long registerMaxDelaySeconds;
    private Thread workThread;

    private final UserAndPassHelper userAndPassHelper;

    private final RegisterExecutor registerExecutor;

    private volatile boolean closed = true;

    public ConnectorConnectionBinding(String kafkaServers, String kafkaConnectAddresses, UserAndPassHelper userAndPassHelper) {
        this(kafkaServers, kafkaConnectAddresses, userAndPassHelper, new ConcurrentHashMap<>(),
                new LinkedBlockingQueue<>(), DEFAULT_DETECTION_MS, REGISTER_FAIL_RETRY_INTERVAL_TIME, DEFAULT_GROUP_AVERAGE_NUMBER, DEFAULT_REGISTER_MAX_DELAY_SECONDS);
    }

    public ConnectorConnectionBinding(String kafkaServers, String kafkaConnectAddresses, UserAndPassHelper userAndPassHelper,
                                      ConcurrentMap<String, MetaData> metaDataMap,
                                      BlockingQueue<String> blockingDeque, Long connectionDetectionTime,
                                      Long registerFailRetryIntervalTime, Integer groupAverageNumber,
                                      Long registerMaxDelaySeconds) {
        this.kafkaServers = kafkaServers;
        this.metaDataMap = metaDataMap;
        this.doRegisterEvent = blockingDeque;
        this.connectionDetectionTime = connectionDetectionTime;
        this.registerFailRetryIntervalTime = registerFailRetryIntervalTime;
        this.groupAverageNumber = groupAverageNumber;
        this.registerMaxDelaySeconds = registerMaxDelaySeconds;

        this.userAndPassHelper = userAndPassHelper;

        this.registerExecutor = new RegisterExecutor(kafkaServers, kafkaConnectAddresses, this);
    }

    public boolean registerInfo(ConnectorInfo connectorInfo) throws SQLException {
        if (isClosed()) {
            return false;
        }
        correctUserNameAndPwd(connectorInfo);

        String name = connectorInfo.getConnectorName();
        synchronized (StringLockUtils.getOrCreateStringLock(name)) {
            genBind(connectorInfo);
        }
        return true;
    }

    private void correctUserNameAndPwd(ConnectorInfo connectorInfo) {
        String userName = connectorInfo.getUserName();
        String password = connectorInfo.getPassword();

        String group = connectorInfo.getGroup();
        String hostName = connectorInfo.getHostName();

        if (userName == null || userName.startsWith(EXPRESSION_PREFIX)) {
            connectorInfo.setUserName(userAndPassHelper.getUserName(group, hostName));
            connectorInfo.setUserNameExpression(userAndPassHelper.getUserNameExpression(group, hostName));
        }
        if (password == null || password.startsWith(EXPRESSION_PREFIX)) {
            connectorInfo.setPassword(userAndPassHelper.getPassWord(group, hostName));
            connectorInfo.setPasswordExpression(userAndPassHelper.getPasswordExpression(group, hostName));
        }
    }

    @Override
    public void run() {
        this.registerExecutor.pullALLConnectors();

        while (true) {
            long maxWaitMs = this.connectionDetectionTime;
            long nowMs = System.currentTimeMillis();

            tryClose();
            register(this.groupAverageNumber, this.registerMaxDelaySeconds);

            long afterDoRegisterMs = System.currentTimeMillis();
            long remainMs = maxWaitMs - (afterDoRegisterMs - nowMs);
            if (remainMs > 0) {
                try {
                    Thread.sleep(remainMs);
                } catch (InterruptedException e) {
                    LOGGER.info("Registration event is coming !!!");
                }
            }

            checkConnectionAndRepair();
            waitWhenNoConnection();
            repairConnectorFailed();
        }
    }

    private void tryClose() {
        synchronized (this) {
            if (this.closed && doRegisterEvent.isEmpty()) {
                this.metaDataMap.clear();
                this.workThread = null;
                notify();
                throw new RuntimeException("ConnectorConnectionBinding Closed safely");
            }
        }
    }

    public void start() {
        if (!isClosed()) {
            return;
        }

        if (this.workThread == null) {
            start0();
            return;
        }
        synchronized (this) {
            if (this.workThread == null) {
                start0();
                return;
            }
            try {
                wait();
            } catch (InterruptedException ignore) {
            }
            start0();

        }

    }

    private void start0() {
        this.closed = false;
        this.workThread = new Thread(this, "ConnectorConnectionBinding-thread-" + threadNum.getAndIncrement());
        workThread.start();
    }

    public void close() {
        if (isClosed()) {
            return;
        }
        this.closed = true;
        if (this.workThread != null) {
            this.workThread.interrupt();
        }

    }

    public boolean isClosed() {
        return this.closed;
    }

    private void repairConnectorFailed() {
        registerExecutor.doRepairConnectorFailed(this.metaDataMap);

    }

    private void genBind(ConnectorInfo connectorInfo) throws SQLException {
        Connection conn = DriverManager.getConnection("jdbc:mysql://" + connectorInfo.getHostName() + ":" + connectorInfo.getPort()
                + "?useUnicode=true&characterEncoding=utf8&user=" + connectorInfo.getUserName() + "&password=" + connectorInfo.getPassword());

        ConnectionWrap connectionWrap = new ConnectionWrap(conn, true, null);
        MetaData metaData = new MetaData(connectionWrap, connectorInfo);

        MetaData previousMetaData = metaDataMap.put(connectorInfo.getConnectorName(), metaData);
        if (previousMetaData != null && !previousMetaData.getConnectionWrap().getConnection().isClosed()) {
            previousMetaData.getConnectionWrap().getConnection().close();
        }

        if (doRegisterEvent.offer(connectorInfo.getGroup())) {
            if (workThread != null) {
                workThread.interrupt();
            }
        } else {
            LOGGER.error("genBind doRegisterEvent already full");
        }

    }

    private void clear(ConnectorInfo connectorInfo) {
        String connectorName = connectorInfo.getConnectorName();
        metaDataMap.remove(connectorName);
        if (!doRegisterEvent.offer(connectorInfo.getGroup())) {
            LOGGER.error("clear doRegisterEvent already full");
        }
    }

    private void register(Integer batchSize, long timeout) {
        List<String> groups = new ArrayList<>(batchSize);
        try {
            Queues.drain(doRegisterEvent, groups, batchSize, timeout, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            this.workThread.interrupt();
        }
        Set<String> groupSets = new HashSet<>(groups);

        Collection<MetaData> fullMetaData = this.metaDataMap.values();

        Map<String, List<MetaData>> fullMetaDataGroupMap =
                fullMetaData.stream().collect(Collectors.groupingBy(MetaData::getGroup));

        groupSets.forEach(group -> {
            List<MetaData> groupMetaData = fullMetaDataGroupMap.get(group);
            Map<String, String> serverIdMap = new HashMap<>();

            if (groupMetaData != null) {
                int size = groupMetaData.size();

                for (int i = size - 1; i >= 0; i--) {
                    MetaData metaData = groupMetaData.get(i);
                    ConnectionWrap connWrap = metaData.getConnectionWrap();
                    if (connWrap == null || !connWrap.getHealthy()) {
                        groupMetaData.remove(metaData);
                    } else {
                        serverIdMap.put(metaData.getConnectorName(), connWrap.genServerId());
                    }
                }
                List<ConnectorInfo> groupConnectorInfos = groupMetaData.stream().map(MetaData::getConnectorInfo).collect(Collectors.toList());
                doRegister(new RegisterGroup(group, this.kafkaServers, groupConnectorInfos, serverIdMap));
            } else {
                doRegister(new RegisterGroup(group, this.kafkaServers, Collections.emptyList(), serverIdMap));
            }
        });

    }

    private void doRegister(RegisterGroup registerGroup) {
        if (!registerExecutor.registerKafkaConnector(registerGroup, this.metaDataMap)) {
            LOGGER.error("group: {} registerKafkaConnector fail", registerGroup.getGroup());
            try {
                Thread.sleep(this.registerFailRetryIntervalTime);
            } catch (InterruptedException e) {
                this.workThread.interrupt();
            }

            if (!doRegisterEvent.offer(registerGroup.getGroup())) {
                LOGGER.error("registerKafkaConnector fail than doRegisterEvent already full");
            }
        }
    }

    private void waitWhenNoConnection() {
        if (this.metaDataMap.size() > 0 || isClosed()) {
            return;
        }
        synchronized (this) {
            try {
                wait();
            } catch (InterruptedException e) {
                LOGGER.info("Registration event is coming !!!");
            }
        }
    }

    private void checkConnectionAndRepair() {
        this.metaDataMap.forEach((connectorName, metaData) -> {
            ConnectionWrap connectionWrap = metaData.getConnectionWrap();
            boolean healthy = connectionWrap.getHealthy();

            if (!healthy) {
                synchronized (StringLockUtils.getOrCreateStringLock(connectorName)) {
                    MetaData newerMetaData = this.metaDataMap.get(connectorName);
                    ConnectionWrap newerConnectionWrap = newerMetaData.getConnectionWrap();
                    boolean newerHealthy = newerConnectionWrap.getHealthy();
                    if (newerHealthy) {
                        return;
                    }
                    LocalDateTime errorTime = newerConnectionWrap.getErrorTime();
                    ConnectorInfo connectorInfo = newerMetaData.getConnectorInfo();
                    try {
                        genBind(connectorInfo);
                    } catch (SQLException e) {
                        LocalDateTime now = LocalDateTime.now();
                        if (now.minusDays(1L).isAfter(errorTime)) {
                            clear(connectorInfo);
                        }
                    }
                    return;
                }
            }

            Connection conn = connectionWrap.getConnection();
            try {
                healthy = conn.isValid(5000) && !conn.isClosed();
            } catch (SQLException ignore) {
            }
            if (!healthy) {
                connectionWrap.setHealthy(false);
                connectionWrap.setErrorTime(LocalDateTime.now());
            }

        });

    }
}
